package com.sfx.shortLink.config;

import com.sfx.shortLink.mq.consumer.ShortLinkStatsSaveConsumer;
import lombok.RequiredArgsConstructor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static com.sfx.shortLink.common.constants.RedisKeyConstant.SHORT_LINK_STATS_STREAM_GROUP_KEY;
import static com.sfx.shortLink.common.constants.RedisKeyConstant.SHORT_LINK_STATS_STREAM_TOPIC_KEY;

/**
 * Redis Stream 消息队列配置
 */
@Configuration
@RequiredArgsConstructor
public class RedisStreamConfiguration {

    private final RedisConnectionFactory redisConnectionFactory;
    private final ShortLinkStatsSaveConsumer shortLinkStatsSaveConsumer;

    @Bean
    public ExecutorService asyncStreamConsumer() {
        AtomicInteger index = new AtomicInteger();
        // 当前的最大cpu核数
        int processors = Runtime.getRuntime().availableProcessors();
        return new ThreadPoolExecutor(processors,
                processors + processors >> 1,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(),
                runnable -> {
                    Thread thread = new Thread(runnable);
                    thread.setName("stream_consumer_short-link_stats_" + index.incrementAndGet());
                    thread.setDaemon(true);
                    return thread;
                }
        );
    }

    // 用于监听redis的Stream中的消息
    @Bean(initMethod = "start", destroyMethod = "stop") // 指定了bean启动和销毁的方法
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(ExecutorService asyncStreamConsumer) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        // 一次最多获取多少条消息
                        .batchSize(10)
                        // 执行从 Stream 拉取到消息的任务流程
                        .executor(asyncStreamConsumer)
                        // 如果没有拉取到消息，需要阻塞的时间。不能大于 ${spring.data.redis.timeout}，否则会超时  --> 类似于sleep 3s
                        .pollTimeout(Duration.ofSeconds(3))
                        .build();

        // 创建了一个用于监听Redis Stream消息的容器，并配置了相应的选项。这个容器可以被用来注册消息监听器，
        // 并在消息到达时执行相应的处理逻辑。在你的代码中，这个容器用于监听短链接统计信息的Redis Stream，
        // 并在有新消息到达时将其传递给shortLinkStatsSaveConsumer进行处理。
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);

        // 用于设置消息自动确认机制。
        streamMessageListenerContainer.receiveAutoAck(
                // 创建了一个消费者，指定了消费者组和消费者名称
                Consumer.from(SHORT_LINK_STATS_STREAM_GROUP_KEY, "stats-consumer"),
                // 创建了一个表示流偏移的对象，指定了要监听的Stream以及从上次消费的位置开始。
                StreamOffset.create(SHORT_LINK_STATS_STREAM_TOPIC_KEY, ReadOffset.lastConsumed()),
                // 指定消息的消费者
                shortLinkStatsSaveConsumer);

        return streamMessageListenerContainer;
    }
}